package demo.db.hbase.utils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;

import demo.db.hbase.vo.CellVO;

public class HbaseUtils {

    private static final Logger logger = LoggerFactory.getLogger(HbaseUtils.class);

    /**
     * 打印Result
     * 
     * @param result
     */
    public static void printResult(Result result) {
        byte[] fimily = Bytes.toBytes("cf1");
        // getNoVersionMap
        NavigableMap<byte[], NavigableMap<byte[], byte[]>> noVersionMap = result.getNoVersionMap();
        logger.info("getNoVersionMap: {}", CommonUtils.parseMap2(noVersionMap));
        // getFamilyMap
        NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(fimily);
        logger.info("getFamilyMap: {}", CommonUtils.parseMap(familyMap));
        // getColumnLatestCell
        Cell cell = result.getColumnLatestCell(fimily, Bytes.toBytes("Name"));
        logger.info("getColumnLatestCell : {}", HbaseUtils.parseCell(cell));
        // getExists
        Boolean exists = result.getExists();
        logger.info("getExists : {}", exists);
        // getMap
        NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> allVersionsMap = result.getMap();
        logger.info("getMap : {}", CommonUtils.parseMap3(allVersionsMap));
        // getRow
        byte[] row = result.getRow();
        logger.info("getRow : {}", Bytes.toString(row));
        // getTotalSizeOfCells
        long totalSizeOfCells = Result.getTotalSizeOfCells(result);
        logger.info("getTotalSizeOfCells : {}", totalSizeOfCells);

        // cellScanner
        try {
            List<CellVO> list = HbaseUtils.parseResult1(result);
            String jsonStr = JSONObject.toJSONString(list);
            logger.info("cellScanner : {}", jsonStr);
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    public static Map<String, String> parseResult2(Result result, byte[] family) throws IOException {
        NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(family);
        return CommonUtils.parseMap(familyMap);
    }

    public static List<CellVO> parseResult1(Result result) throws IOException {
        List<CellVO> list = new ArrayList<>();
        CellScanner cellScanner = result.cellScanner();
        while (cellScanner.advance()) {
            Cell cell = cellScanner.current();
            CellVO vo = HbaseUtils.parseCell(cell);
            list.add(vo);
        }
        return list;
    }

    public static CellVO parseCell(Cell cell) {
        if (cell == null) {
            return null;
        }
        byte[] familyArray = cell.getFamilyArray();
        String family = Bytes.toString(familyArray, cell.getFamilyOffset(), cell.getFamilyLength());

        byte[] qualifierArray = cell.getQualifierArray();
        String qualifier = Bytes.toString(qualifierArray, cell.getQualifierOffset(), cell.getQualifierLength());

        long sequenceId = cell.getSequenceId();

        byte[] rowArray = cell.getRowArray();
        String row = Bytes.toString(rowArray, cell.getRowOffset(), cell.getRowLength());

        byte[] tagsArray = cell.getTagsArray();
        String tags = Bytes.toString(tagsArray, cell.getTagsOffset(), cell.getTagsLength());

        long timestamp = cell.getTimestamp();

        byte[] valueArray = cell.getValueArray();
        String value = Bytes.toString(valueArray, cell.getValueOffset(), cell.getValueLength());

        KeyValue.Type type = KeyValue.Type.codeToType(cell.getTypeByte());

        CellVO vo = new CellVO();
        vo.setFamily(family);
        vo.setQualifier(qualifier);
        vo.setRow(row);
        vo.setSequenceId(sequenceId);
        vo.setTags(tags);
        vo.setTimestamp(timestamp);
        vo.setType(type);
        vo.setValue(value);
        return vo;
    }
}
